package com.moli.iov.async;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <p>
 *  异步管道
 *  既是管道，也是业务
 *
 *  适合需要排队异步执行任务，
 *  如果需要创建大量的Pipeline，需要把ExecutorService 初始化移出，由外部提供
 *
 * </p>
 *
 * @Auther: zql
 * @Date: 2018/12/5 17:59
 */
public abstract class AsyncPipeline<D> implements AutoCloseable {

    protected static final Logger logger = LoggerFactory.getLogger(AsyncPipeline.class);
    private BlockingQueue<D> blockingQueue;
    private ThreadFactory threadFactory = Executors.defaultThreadFactory();
    private int queueSize = 1024 * 5;
    private volatile boolean isBreak = false;

    /**
     * 业务
     * @param paramD
     * @throws Exception
     */
    protected abstract void onEvent(D paramD);

    /**
     * 根据指定的队列长度，初始化异步管道
     * @param queueSize
     */
    public final void init(int queueSize) {
        if(queueSize > 0){
            this.queueSize = queueSize;
        }
        init();
    }

    /**
     * 初始化异步管道
     */
    @PostConstruct
    public final void init(){
        this.isBreak = false;
        logger.info("init [{}] queueSize:{} begin...",  getClass(), queueSize);

        this.blockingQueue = new LinkedBlockingQueue(queueSize);

        threadFactory.newThread(()-> {

            while (!AsyncPipeline.this.isBreak) {
                Object data = null;
                try {
                    data = AsyncPipeline.this.blockingQueue.take();
                } catch (Exception e) {
                    logger.error("AsyncPipeline [{}] take Error:{}",  getClass(), e.getMessage(), e);
                }

                try {
                    if (data != null){
                        AsyncPipeline.this.onEvent((D) data);
                    }
                } catch (Exception e) {
                    logger.error("AsyncPipeline [{}] onEvent Error:{}",  getClass(), e.getMessage(), e);
                }
            }
        }).start();

        logger.info("init [{}] end.", getClass());
    }

    /**
     * 添加数据
     * @param data
     * @throws InterruptedException
     */
    public final void onData(D data) throws InterruptedException {
        try {
            this.blockingQueue.put(data);
        } catch (Exception e) {
            logger.error("AsyncPipeline [{}] onData Error:{}",  getClass(), e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 队列元素个数
     * @return
     */
    public int getTodoSize() {
        return this.blockingQueue.size();
    }


    /**
     * 队列长度
     * @return
     */
    public int getPipelineSize(){
        return this.queueSize;
    }

    /**
     * 清空队列
     */
    public void clear() {
        this.blockingQueue.clear();
    }


    /**
     * 钩子
     */
    @Override
    public final void close() {
        logger.debug("close AsyncPipeline [{}] begin", getClass());
        this.isBreak = true;
        this.blockingQueue.clear();
        logger.debug("close AsyncPipeline [{}] end", getClass());
    }
}
